/*
package com.etc.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import scala.Tuple2;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

*/
/**
 * @author 杰哥
 * @Title: StreamingKafka
 * @ProjectName scalaworldcount
 * @Description: TODO
 * @date 2019/8/169:53
 *//*

public class StreamingKafka {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("StreamingKafka").setMaster("local[*]");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(1));
        Map<String, Integer> map = new HashMap<String, Integer>();
        map.put("WordCount", 1);
        JavaPairReceiverInputDStream<String, String> line = KafkaUtils.createStream(jsc, "192.168.88.5:2181,192.168.88.6:2181,192.168.88.7:2181", "DefaultConsumerGroup", map);
        JavaDStream<String> stringJavaDStream = line.flatMap(new FlatMapFunction<Tuple2<String, String>, String>() {
            @Override
            public Iterator<String> call(Tuple2<String, String> stringStringTuple2) throws Exception {
                return Arrays.asList(stringStringTuple2._2.split(" ")).iterator();
            }
        });
        JavaPairDStream<String, Integer> stringIntegerJavaPairDStream = stringJavaDStream.mapToPair(

                new PairFunction<String, String, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Integer> call(String word)
                            throws Exception {
                        return new Tuple2<String, Integer>(word, 1);
                    }

                });

        JavaPairDStream<String, Integer> wordCounts = stringIntegerJavaPairDStream.reduceByKey(

                new Function2<Integer, Integer, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Integer call(Integer v1, Integer v2) throws Exception {
                        return v1 + v2;
                    }

                });
        wordCounts.print();

        jsc.start();
        try{
        jsc.awaitTermination();
        }catch (Exception e){
            e.printStackTrace();
        }
        jsc.close();
    }
}
*/
